package com.bigdata.flink.conf;

import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

public class KafkaProperties {

    private static String topic = ConfigurationManager.getProperty("topic");

    private static int partition = ConfigurationManager.getInteger("numOfPartition");

    private static short replication = ConfigurationManager
            .getInteger("numOfReplication")
            .shortValue();

    public static Properties getKafkaProperties(){
        Properties prop = new Properties();
        prop.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, ConfigurationManager.getProperty("bootstrap.servers"));
        prop.put(ConsumerConfig.GROUP_ID_CONFIG, ConfigurationManager.getProperty("group.id"));
        prop.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, ConfigurationManager.getProperty("enable.auto.commit"));
        prop.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        prop.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        //prop.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"earliest");
        prop.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG,"1000");
        return prop;
    }

    public static Producer<String, String> createProducer(){
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, ConfigurationManager.getProperty("bootstrap.servers"));
        properties.put(ProducerConfig.CLIENT_ID_CONFIG, ConfigurationManager.getProperty("client.id"));
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        return new KafkaProducer<>(properties);
    }

    public static AdminClient createAdminClient(){
        Properties properties = new Properties();
        properties.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG,
                ConfigurationManager.getProperty("bootstrap.servers"));
        AdminClient admin = AdminClient.create(properties);
        return admin;
    }

    public static String getTopic(){
        return topic;
    }

    public static int getPartition() {
        return partition;
    }

    public static short getReplication() {
        return replication;
    }

}
